package io.kiki.sba.registry.server.clientInterface.store;


import io.kiki.sba.registry.common.model.ClientToClientInterfaceServerNodeId;
import io.kiki.sba.registry.common.model.sessionserver.PubSubDataInfoIdResp;
import io.kiki.sba.registry.common.model.store.Publisher;
import io.kiki.sba.registry.common.model.store.Subscriber;
import io.kiki.sba.registry.server.clientInterface.connections.ConnectionsService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.Map.Entry;
import java.util.stream.Collectors;

public class FetchPubSubDataInfoIdService {

    /**
     * store subscribers
     */
    @Autowired
    protected Subscribers sessionSubscribers;
    /**
     * store publishers
     */
    @Autowired
    protected Publishers sessionPublishers;
    @Autowired
    private ConnectionsService connectionsService;

    public PubSubDataInfoIdResp queryByIps(Set<String> ips) {

        PubSubDataInfoIdResp resp = new PubSubDataInfoIdResp();
        List<ClientToClientInterfaceServerNodeId> clientToClientInterfaceServerNodeIds = connectionsService.getIpConnects(ips);
        if (CollectionUtils.isEmpty(clientToClientInterfaceServerNodeIds)) {
            return resp;
        }

        Map<ClientToClientInterfaceServerNodeId, Map<String, Publisher>> connectIdPubMap = sessionPublishers.queryByConnectIds(new HashSet<>(clientToClientInterfaceServerNodeIds));
        Map<ClientToClientInterfaceServerNodeId, Map<String, Subscriber>> connectIdSubMap = sessionSubscribers.queryByConnectIds(new HashSet<>(clientToClientInterfaceServerNodeIds));

        // collect pub and sub dataInfoIds
        Map<String, Set<String>> pubs = parsePubDataInfoIds(connectIdPubMap);
        Map<String, Set<String>> subs = parseSubDataInfoIds(connectIdSubMap);

        resp.setPubDataInfoIds(pubs);
        resp.setSubDataInfoIds(subs);
        return resp;
    }

    protected Map<String, Set<String>> parsePubDataInfoIds(Map<ClientToClientInterfaceServerNodeId, Map<String, Publisher>> connectIdPubMap) {
        Map<String, Set<String>> pubs = new HashMap<>(connectIdPubMap.size());
        for (Entry<ClientToClientInterfaceServerNodeId, Map<String, Publisher>> pubEntry : connectIdPubMap.entrySet()) {
            if (CollectionUtils.isEmpty(pubEntry.getValue())) {
                continue;
            }
            Set<String> set = pubs.computeIfAbsent(pubEntry.getKey().getClientIp(), k -> new HashSet<>());
            Set<String> collect = pubEntry.getValue().values().stream().map(Publisher::getDataInfoId).collect(Collectors.toSet());
            set.addAll(collect);
        }
        return pubs;
    }

    protected Map<String, Set<String>> parseSubDataInfoIds(Map<ClientToClientInterfaceServerNodeId, Map<String, Subscriber>> connectIdSubMap) {
        Map<String, Set<String>> subs = new HashMap<>(connectIdSubMap.size());
        for (Entry<ClientToClientInterfaceServerNodeId, Map<String, Subscriber>> subEntry : connectIdSubMap.entrySet()) {
            if (CollectionUtils.isEmpty(subEntry.getValue())) {
                continue;
            }
            Set<String> set = subs.computeIfAbsent(subEntry.getKey().getClientIp(), k -> new HashSet<>());
            Set<String> collect = subEntry.getValue().values().stream().map(Subscriber::getDataInfoId).collect(Collectors.toSet());
            set.addAll(collect);
        }
        return subs;
    }
}
